#include "Socket.h"

std::string SocketWorker::GetRemoteAddr()
{
	return remote_addr;
}


int SocketWorker::Stop()
{
	m_bStop = true;

	ACE_Message_Block * mb = NULL;
	ACE_NEW_RETURN(mb , ACE_Message_Block(1) , -1); 
	mb->copy("0" , 1);
	this->putq(mb);

	this->wait();
	return 0;
}



int SocketWorker::svc(void)
{

	ACE_Message_Block * mb = NULL;

	while(!m_bStop)
	{
		if(this->getq(mb , NULL) != -1) 
		{
			Execute(mb->rd_ptr());
			mb->release();
		}
	}


	return 0;
}


int SocketWorker::Send(PACKET_BASE& packet)
{

	std::string snd_text = packet.ToString();
	int len = snd_text.size();

	int bufsize = len + LENGTH_SIZE ;

	BYTE tmp[MAX_BUFFER + 1];
	memset(tmp , 0x00 ,  MAX_BUFFER + 1);
	
	int a = htonl(len);	
	memcpy(tmp , &a , LENGTH_SIZE);
	memcpy(tmp + LENGTH_SIZE , snd_text.c_str() , len);

	m_SendLock.acquire();
	int sendsize = m_Stream.send_n(tmp , bufsize);
	m_SendLock.release();

	m_SendTime = time(NULL);

	return sendsize;

}

ACE_HANDLE SocketWorker::get_handle(void) const 
{
    return m_Stream.get_handle();
}

int SocketWorker::handle_input (ACE_HANDLE fd)
{
	m_nCheckCount = 0;

	BYTE	buffer[MAX_BUFFER + 1];
	memset(buffer , 0x00 , MAX_BUFFER + 1);

	ssize_t bytes_received = m_Stream.recv (buffer, MAX_BUFFER);

	if(bytes_received <= 0)
		return -1;

	memcpy(m_SaveBuffer + m_nSaveSize , buffer , bytes_received);
	m_nSaveSize += bytes_received;

	while(m_nSaveSize > LENGTH_SIZE)
	{
		int len;
		memcpy(&len , m_SaveBuffer , LENGTH_SIZE);
		int packet_len = ntohl(len);



		if(packet_len + LENGTH_SIZE <= m_nSaveSize)
		{

			memset(buffer , 0x00 , sizeof(buffer));
			memcpy(buffer , m_SaveBuffer + LENGTH_SIZE , packet_len); 


			ACE_Message_Block * mb = NULL;
			ACE_NEW_RETURN(mb , ACE_Message_Block(packet_len) , -1); 
			mb->copy((const char*)buffer , packet_len);
			this->putq(mb);

			int move_offset = LENGTH_SIZE + packet_len;
			m_nSaveSize -= move_offset;									
			ACE_ASSERT(m_nSaveSize >= 0);

			if(m_nSaveSize > 0)
			{
				memcpy(buffer , m_SaveBuffer + move_offset , m_nSaveSize );  
				memcpy(m_SaveBuffer , buffer , m_nSaveSize);					
			}
		}
		else
			break;

	}

	return 0;

};


int SocketWorker::handle_timeout(const ACE_Time_Value &current_time, const void *act)
{
	CheckConnection();
	return 0;
}


int SocketWorker::CheckConnection()
{
	if(++m_nCheckCount >= Settings::GetSettings()->CONNECTION_CHECK_COUNT)
	{

	
#ifdef CONTROLLER 
		if(_STOPPED != state)
#else	
		if('k' != Settings::GetSettings()->OPERATION_MODE && 'K' != Settings::GetSettings()->OPERATION_MODE)
#endif
		{
			printf("%s : abnormal connection timeout\n" , this->GetRemoteAddr().c_str());
			LOG::PushProcessLog(_ERROR , this->GetRemoteAddr() + " : abnormal connection timeout");
		}



		timeout = true;

		return this->m_Stream.close();
	}
	else
	{

		if(Settings::GetSettings()->CONNECTION_CHECK_DELAY <= (time(NULL) - m_SendTime))
		{
			PACKET_TICK tick;
			Send(tick);
		}
	}
	return 0;
}